package com.heaven.service.minio;

import com.heaven.customerServer.customerController.vo.uploadFile.*;
import com.heaven.constant.BigFileConstant;
import com.heaven.event.UpToMinioEvent;
import com.heaven.exception.BizCodeEnum;
import com.heaven.exception.RRException;
import io.minio.*;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.io.*;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 大文件存储在minio中为分块存储，需要时将分块传输给客户端，客户端将这些分块拼接起来就可以
 * 大文件分块临时存储格式：
 *  {COMMON_PREFIX}/{md5第一位字符}/{md5第二位字符}/{md5}/blobs/{md5}--{1,2,3...}.suffix
 * 大文件上传完毕后就存储最后一个分块来标记大文件已经上传完毕, 大文件完整存储格式：:
 *  {COMMON_PREFIX}/{md5第一位字符}/{md5第二位字符}/{md5}/{md5}.suffix
 * 标记文件格式: {COMMON_PREFIX}/exists/{md5}
 */
@Service
@Slf4j
public class BigFileUploadService extends MinIoService {
    @Resource
    private MinioClient minioClient;
    @Resource
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
    @Resource
    private ApplicationEventPublisher eventPublisher;

    @Value("${minio.bucket1}")
    public String bucketName;

    /** 本地缓存文件夹 */
    @Value("${local.store-path}")
    public String LOCAL_STORE_PATH;

    @Value("${server.model:stand-alone}")
    private String SERVER_MODEL;

    /**
     * 创建大文件分块的暂存文件夹
     * @param md5 大文件MD5
     */
    private void createBlobFile(String md5){
        switch (SERVER_MODEL) {
            case "stand-alone":
                // 在本地创建大文件分块临时存储文件夹
                String blobDir = BigFileConstant.getBlobPathPrefix(md5);
                Path path = Paths.get(LOCAL_STORE_PATH, blobDir);
                path.toFile().mkdirs();
                break;
            case "cluster":
                // 在minio上创建大文件分块临时存储文件夹
                String dir = BigFileConstant.getBlobPathPrefix(md5);
                try {
                    ObjectWriteResponse objectWriteResponse = minioClient.putObject(
                            PutObjectArgs.builder().bucket(bucketName).object(dir).stream(
                                            new ByteArrayInputStream(new byte[]{}), 0, -1)
                                    .build());
                    if(objectWriteResponse != null) {
                        log.info("{} temp blob file created. tag: {}", md5, objectWriteResponse.etag());
                    }
                }catch (Exception e) {
                    log.error("create blob temp file failed !!!");
                    throw new RRException("创建 blob 暂存文件失败");
                }
                break;
            default:
                throw new RRException("暂不支持的服务器模式");
        }
    }

    /**
     * 在minio上创建大文件存在的标记文件
     * @param md5 大文件MD5值
     * @param is 标记输入流 1B
     * @param contentType 标记文件类型，默认为 application/octet-stream
     */
    private void createBigExistFlag(String md5, InputStream is, String contentType) {
        try {
            int len = 1;
            byte []flagBuffer = new byte[len];
            is.read(flagBuffer);
            InputStream flagIs = new ByteArrayInputStream(flagBuffer);
            putObject(bucketName, BigFileConstant.getBigFileFlagObjectName(md5),flagIs,len, contentType);
            flagIs.close();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RRException("创建大文件存在标记异常-"+e.getMessage());
        }

    }

    /**
     * 检验大文件是否完整存在(条件是文件整体存储在minio中)
     * @param md5 大文件MD5
     * @return 完整存在:true, 不完整存在:false
     */
    public boolean isBigFileWholeExist(String md5, String fileName) {
        String bigFileFlagObject = BigFileConstant.getBigFileObjectName(md5, BigFileConstant.getFileNameSuffix(fileName));
        return isObjectExist(bucketName, bigFileFlagObject);
    }

    /**
     * 检验存储文件分块的父文件夹是否存在
     * @param md5 文件MD5
     * @return 存在:true, 不存在:false
     */
    public boolean isBlobDirExist(String md5) {
        String blobPathPrefix = BigFileConstant.getBlobPathPrefix(md5);
        return switch (SERVER_MODEL) {
            case "stand-alone" -> Paths.get(LOCAL_STORE_PATH, blobPathPrefix).toFile().exists();
            case "cluster" -> isDirExist(bucketName, blobPathPrefix);
            default -> throw new RRException("暂不支持的服务器模式");
        };
    }

    /**
     * 检验存储文件分块是否存在
     * @param md5 文件MD5
     * @param blobIndex 文件分块索引
     * @param fileSuffix 文件后缀
     * @return 存在:true, 不存在:false
     */
    public boolean isBlobExist(String md5, long blobIndex, String fileSuffix) {
        String blobObjectName = BigFileConstant.getBlobObjectName(md5, blobIndex, fileSuffix);
        return switch (SERVER_MODEL) {
            case "stand-alone" ->
                // 分块存储在本地
                Paths.get(LOCAL_STORE_PATH, blobObjectName).toFile().exists();
            case "cluster" ->
                // 分块存储在minio
                isObjectExist(bucketName, blobObjectName);
            default -> throw new RRException("暂不支持的服务器模式");
        };
    }

    /**
     * 检验大文件分块是否完整
     * @param md5 文件md5
     * @param fileSuffix 文件后缀
     * @param blobNum 文件分块数量
     * @return 完整存在:true, 不完整存在:false
     */
    private boolean verifyFile(String md5, String fileSuffix, long blobNum) {
        // 验证文件完整性
        String blobsPathPrefix = BigFileConstant.getBlobPathPrefix(md5);
        log.info("begin to verify the integrity of {}...", blobsPathPrefix);
        AtomicBoolean flag = new AtomicBoolean(true);
        List<CompletableFuture<Void>> tasks = new ArrayList<>();
        for (long i = 0; i < blobNum; i++) {
            // 开始异步操作
            long finalI = i;
            CompletableFuture<Void> task = CompletableFuture.runAsync(() -> {
                log.info("{} verify progress {}/{}", blobsPathPrefix, finalI + 1, blobNum);
                boolean objectExist = isBlobExist(md5, finalI, fileSuffix);
                if (!objectExist) {
                    flag.set(false);
                }
            }, threadPoolTaskExecutor);
            tasks.add(task);
            if (tasks.size() >= 10) {
                blockTask(tasks);
            }
        }
        if(!tasks.isEmpty()) blockTask(tasks);

        String bigFileObject = BigFileConstant.getBlobPathPrefix(md5);
        log.info("{} verify completed!", bigFileObject);
        return flag.get();
    }

    /**
     * 合并本地的md5分块文件
     * @param targetFilePath 目标文件路径
     * @param md5 文件MD5
     * @param blobNum 文件分块数量
     * @param fileSuffix 文件后缀
     */
    private void mergeLocalCache(Path targetFilePath ,String md5, long blobNum, String fileSuffix) {
        if(!targetFilePath.getParent().toFile().exists()) {
            // 创建父目录
            targetFilePath.getParent().toFile().mkdirs();
        }
        File targetFile = targetFilePath.toFile();
        try {
            RandomAccessFile raf_rw = new RandomAccessFile(targetFile, "rwd");
            byte [] buffer = new byte[1024];
            for (int i = 0; i < blobNum; i++) {
                String localBlobChild = BigFileConstant.getBlobObjectName(md5, i, fileSuffix);
                File blobFile = Paths.get(LOCAL_STORE_PATH, localBlobChild).toFile();
                int len;
                RandomAccessFile raf_r = new RandomAccessFile(blobFile, "r");
                while ((len = raf_r.read(buffer)) != -1) {
                    raf_rw.write(buffer, 0, len);
                }
                raf_r.close();
            }
            raf_rw.close();
            log.info("merge local blob to file: {} completed!", targetFile);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 获取大文件上传的起始分块索引
     * @param md5 大文件MD5
     *            maxBlobSize 分块大小
     *            size 大文件大小
     *            fileName 大文件名称
     * @return BlobStartRes
     */
    public BlobStartRes getStartBlobIndex(String md5, Long maxBlobSize, Long size, String fileName) {
        if(md5 == null || maxBlobSize == null || size == null || fileName == null
            || md5.isEmpty() || size == 0L || fileName.isEmpty()) {
            throw new RRException("参数错误");
        }

        //1. 检验大文件是否存在
        boolean isExist = isBigFileWholeExist(md5, fileName);
        try {
            if(isExist){
                log.warn("{} is already existed !", md5);
                String bigFileObjectName = BigFileConstant.getBigFileObjectName(md5, BigFileConstant.getFileNameSuffix(fileName));
                return new BlobStartRes(true,-1, bucketName, bigFileObjectName);
            }

            //2. 大文件不存在的情况下，检验大文件分块文件夹是否存在
            boolean flag = isBlobDirExist(md5);
            if(!flag) {
                //2.1 大文件分块不存在：创建一个文件夹来存储该大文件的分块，返回{isExist： false, index: 0, blobPath: {dir}}
                createBlobFile(md5);
                return new BlobStartRes(false, 0);
            } else {
                //2.2 大文件分块存在：返回{isExist: false, index: index-1, blobPath: {dir}}
                long blobNum = (long) Math.ceil(size *1.0 / maxBlobSize);
                String fileSuffix = BigFileConstant.getFileNameSuffix(fileName);
                long l = 0, r = blobNum-1, m;
                while (l <= r) {
                    m = (l+r)/2;
                    if (!isBlobExist(md5, m, fileSuffix)) {
                        if(m == 0) {
                            return new BlobStartRes(false, 0);
                        } else {
                            if(isBlobExist(md5, m-1, fileSuffix)){
                                return new BlobStartRes(false, m-1);
                            } else {
                                r = m-1;
                            }
                        }
                    } else {
                        l = m+1;
                    }
                }
                //所有分块集齐, 等待合并
                return new BlobStartRes(false, blobNum-1);
            }
        } catch (Exception e) {
            throw new RRException("获取分块上传文件索引失败");
        }
    }

    /**
     * 上传文件分块到minio
     * @param blobIs 文件分块输入流
     * @param blobSize 文件分块实际大小
     * @param feature 大文件基本信息
     * @return 上传反馈体
     */
    //上传大文件的分块
    public UploadBigFileRes uploadBigFileBlob(InputStream blobIs, long blobSize, BigFileFeatureReq feature) {
        String fileSuffix = BigFileConstant.getFileNameSuffix(feature.getFileName());
        // 1.分块写入到存储文件分块文件夹{dir}
        String blobObject = BigFileConstant.getBlobObjectName(feature.getMd5(), feature.getIndex(), fileSuffix);
        log.info("up blob {} to minio server...", blobObject);
        putObject(bucketName, blobObject, blobIs, blobSize,"application/octet-stream");

        // 2.判断是否是最后一个分块文件
        long blobNum = (long) Math.ceil(feature.getSize() *1.0 / feature.getMaxBlobSize());
        String bigFileObject = BigFileConstant.getBigFileObjectName(feature.getMd5(), fileSuffix);
        log.info("{} loading {}/{} ...", bigFileObject, feature.getIndex(), blobNum);
        if (blobNum -1 == feature.getIndex()) {
            // 2.1 是最后一个分块，给客户端返回上传完成标志
            //createBigExistFlag(feature.getMd5(), blobIs, "application/octet-stream");
            return new UploadBigFileRes(1, bucketName, bigFileObject);
        } else {
            // 2.2 不是最后一个分块，返回未完成上传的反馈
            return new UploadBigFileRes(0, bucketName, bigFileObject);
        }
    }

    /**
     * 阻塞异步任务直至全部执行完成, 并清除掉任务列表的所有任务
     * @param tasks 没有返回值的异步任务
     */
    private void blockTask(List<CompletableFuture<Void>> tasks) {
        CompletableFuture<Void> combine = CompletableFuture.allOf(tasks.toArray(new CompletableFuture[0]));
        try {
            combine.get();
            log.info("all tasks completed!");
            tasks.clear();
        } catch (InterruptedException | ExecutionException e) {
            log.error(BizCodeEnum.THREAD_EXCEPTION.getMsg()+"合并线程任务异常");
            if(e instanceof InterruptedException){
                Thread.currentThread().interrupt();
                throw new RRException(BizCodeEnum.THREAD_EXCEPTION, "线程被中断", e);
            } else {
                Throwable cause = e.getCause();
                throw new RRException(BizCodeEnum.THREAD_EXCEPTION, "线程执行过程中的异常", cause);
            }
        }
    }


    /**
     * 合并分块文件
     * @param md5 文件md5
     * @param maxBlobSize 文件分块大小
     * @param fileSize 文件大小（单位: B）
     * @param fileName 文件名
     */
    public void mergeBlobs(String md5, long maxBlobSize, long fileSize, String fileName) {
        CompletableFuture.runAsync(()->{
            String fileSuffix = BigFileConstant.getFileNameSuffix(fileName);
            long blobNum = (long) Math.ceil(fileSize *1.0 / maxBlobSize);
            String fileObjectName = BigFileConstant.getBigFileObjectName(md5, fileSuffix);
            boolean flag = verifyFile(md5, fileSuffix, blobNum);
            if(flag) {
                log.info("begin to merge blobs: {} -> {}", BigFileConstant.getBlobPathPrefix(md5), fileObjectName);
                List<CompletableFuture<Void>> tasks = new ArrayList<>();
                // 将分块保存到本地
                for (long i = 0; i < blobNum; i++) {
                    long finalI = i;
                    CompletableFuture<Void> task = CompletableFuture.runAsync(() -> {
                        String blobObjectName = BigFileConstant.getBlobObjectName(md5, finalI, fileSuffix);
                        try {
                            Path blob = Paths.get(LOCAL_STORE_PATH, blobObjectName);
                            if (!blob.getParent().toFile().exists()) {
                                // 创建父目录
                                blob.getParent().toFile().mkdirs();
                            }
                            // 将分块存储到服务器本地
                            minioClient.downloadObject(
                                    DownloadObjectArgs.builder()
                                            .bucket(bucketName)
                                            .object(blobObjectName)
                                            .filename(blob.toString())
                                            .build());
                            log.info("download blob{}/{}... to {}", finalI, blobNum, fileObjectName);
                        } catch (Exception e) {
                            throw new RRException("获取 minio 文件流异常-", e);
                        }
                    }, threadPoolTaskExecutor);
                    tasks.add(task);

                    if (tasks.size() >= 200) {
                        blockTask(tasks);
                    }
                }
                if (!tasks.isEmpty()) {
                    blockTask(tasks);
                }

                // 将服务器本地的分块合并
                log.info("cache {} download completed! begin to put it to merge...", fileObjectName);
                Path targetFilePath = Paths.get(LOCAL_STORE_PATH, BigFileConstant.getBigFileObjectName(md5, fileSuffix));
                mergeLocalCache(targetFilePath, md5, blobNum, fileSuffix);

                // 将合并后的文件上传到服务器
                uploadLocalFile(bucketName,fileObjectName, targetFilePath.toString());

                // 发布本地事件
                eventPublisher.publishEvent(new UpToMinioEvent(this, md5, fileObjectName));

                // 删除minio缓存的分块
                String bigFileBlobsPath = BigFileConstant.getBlobPathPrefix(md5);
                delDir(bucketName, bigFileBlobsPath);

                //删除本地缓存的分块
                try {
                    FileUtils.deleteDirectory(Paths.get(LOCAL_STORE_PATH, BigFileConstant.bigFilePathPrefix(md5)).toFile());
                } catch (IOException e) {
                    throw new RRException("删除本地缓存文件异常", e);
                }

            } else {
                log.error("{} is incomplete, can't merge!", fileObjectName);
                // TODO  补偿方案
            }
        }, threadPoolTaskExecutor);
    }


    /**
     * 单次上传一个文件
     * @param feature 文件基本信息
     * @param inputStream 文件流
     * @return 反馈体
     */
    public UploadFileRes uploadSingleWhole(FileFeatureReq feature, InputStream inputStream) {
        String fileSuffix = BigFileConstant.getFileNameSuffix(feature.getFileName());
        String fileObjectName = BigFileConstant.getBigFileObjectName(feature.getMd5(), fileSuffix);
        boolean flag = isObjectExist(bucketName, fileObjectName);
        if(!flag) {
            putObject(bucketName, fileObjectName, inputStream, feature.getSize(), "application/octet-stream");
            // 发布本地事件
            eventPublisher.publishEvent(new UpToMinioEvent(this, feature.getMd5(), fileObjectName));
        }
        return new UploadFileRes(1, bucketName, fileObjectName);
    }

    /**
     * 上传文件分块到本地
     * @param blobIs 文件分块流
     * @param feature 大文件基本信息
     * @return 上传反馈体
     */
    public UploadBigFileRes uploadFileBlobLocalCache(InputStream blobIs, BigFileFeatureReq feature) {
        String fileSuffix = BigFileConstant.getFileNameSuffix(feature.getFileName());
        // 1.分块写入到存储文件分块文件夹{dir}
        String blobObject = BigFileConstant.getBlobObjectName(feature.getMd5(), feature.getIndex(), fileSuffix);
        log.info("write blob {} to local...", blobObject);
        File blobFile = Paths.get(LOCAL_STORE_PATH, blobObject).toFile();
        try (OutputStream os = new FileOutputStream(blobFile)) {
            IOUtils.copy(blobIs, os);
        } catch (IOException e) {
            throw new RRException(BizCodeEnum.FILE_IO_EXCEPTION, "写文件异常",e);
        }

        // 2.判断是否是最后一个分块文件
        long blobNum = (long) Math.ceil(feature.getSize() *1.0 / feature.getMaxBlobSize());
        String bigFileObject = BigFileConstant.getBigFileObjectName(feature.getMd5(), fileSuffix);
        log.info("{} loading {}/{} ...", bigFileObject, feature.getIndex(), blobNum);
        if (blobNum -1 == feature.getIndex()) {
            // 2.1 是最后一个分块，给客户端返回上传完成标志
            //createBigExistFlag(feature.getMd5(), blobIs, "application/octet-stream");
            return new UploadBigFileRes(1, bucketName, bigFileObject);
        } else {
            // 2.2 不是最后一个分块，返回未完成上传的反馈
            return new UploadBigFileRes(0, bucketName, bigFileObject);
        }
    }

    public void mergeBlobsLocalCache(String md5, long maxBlobSize, Long size, String fileName) {
        CompletableFuture.runAsync(()->{
            String fileSuffix = BigFileConstant.getFileNameSuffix(fileName);
            long blobNum = (long) Math.ceil(size *1.0 / maxBlobSize);
            String fileObjectName = BigFileConstant.getBigFileObjectName(md5, fileSuffix);
            boolean flag = verifyFile(md5, fileSuffix, blobNum);
            if(flag) {
                log.info("begin to merge blobs: {} -> {}", BigFileConstant.getBlobPathPrefix(md5), fileObjectName);
                // 将服务器本地的分块合并
                Path targetFilePath = Paths.get(LOCAL_STORE_PATH, BigFileConstant.getBigFileObjectName(md5, fileSuffix));
                mergeLocalCache(targetFilePath, md5, blobNum, fileSuffix);
                // 将合并后的文件上传到服务器
                uploadLocalFile(bucketName,fileObjectName, targetFilePath.toString());

                // 发布本地事件
                eventPublisher.publishEvent(new UpToMinioEvent(this, md5, fileObjectName));

                //删除本地缓存的分块
                try {
                    FileUtils.deleteDirectory(Paths.get(LOCAL_STORE_PATH, BigFileConstant.bigFilePathPrefix(md5)).toFile());
                } catch (IOException e) {
                    throw new RRException("删除本地缓存文件异常", e);
                }

            } else {
                log.error("{} is incomplete, can't merge!", fileObjectName);
                // TODO  补偿方案
            }
        }, threadPoolTaskExecutor);
    }
}
